package com.atcyj.gmall.realtime.dwd.db.split.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atcyj.gamll.realtime.common.base.BaseApp;
import com.atcyj.gamll.realtime.common.constant.Constant;
import com.atcyj.gamll.realtime.common.util.DateFormatUtil;
import com.atcyj.gamll.realtime.common.util.FlinkSinkUtil;
import io.debezium.data.Json;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * @author: cyj
 * @date: 2024/8/16
 */
@Slf4j
public class DwdBaseLog extends BaseApp {
    public static void main(String[] args) {
        new DwdBaseLog().start(10011, 4, "dwd_base_app", Constant.TOPIC_LOG);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        // 核心业务处理
        // TODO 1. 数据清洗加转换
        SingleOutputStreamOperator<JSONObject> jsonObjStream = etl(stream);


        // TODO 2. 进行新旧访客修复
        // 需要使用keyBy,需要注册水位线
        // 注意：1. 水位线策提取的时间戳不可以为空
        //      2. keyBy时的key不可以为空
        KeyedStream<JSONObject, String> keyedStream = getKeyedStreamWithWatermarks(jsonObjStream);

        SingleOutputStreamOperator<JSONObject> isNewFixedStream = isNewFix(keyedStream);


        // TODO 3. 用侧输出流拆分日志
        // start、page、action、display、err

        // new 对象的时候，无法传递泛型，泛型擦除问题
        // OutputTag<String> startTag = new OutputTag<String>("start") { };
        OutputTag<String> startTag = new OutputTag<String>("start", TypeInformation.of(String.class));
        OutputTag<String> actionTag = new OutputTag<String>("action", TypeInformation.of(String.class));
        OutputTag<String> displayTag = new OutputTag<String>("display", TypeInformation.of(String.class));
        OutputTag<String> errorTag = new OutputTag<String>("error", TypeInformation.of(String.class));


        SingleOutputStreamOperator<String> pageStream = splitLog(isNewFixedStream, startTag, actionTag, displayTag, errorTag);

        SideOutputDataStream<String> startStream = pageStream.getSideOutput(startTag);
        SideOutputDataStream<String> errorStream = pageStream.getSideOutput(errorTag);
        SideOutputDataStream<String> actionStream = pageStream.getSideOutput(actionTag);
        SideOutputDataStream<String> displayStream = pageStream.getSideOutput(displayTag);

        // TODO 4.  写出到kafka中
        pageStream.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
        startStream.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        errorStream.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        actionStream.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
        displayStream.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));


    }

    private SingleOutputStreamOperator<String> splitLog(SingleOutputStreamOperator<JSONObject> isNewFixedStream, OutputTag<String> startTag, OutputTag<String> actionTag, OutputTag<String> displayTag, OutputTag<String> errorTag) {
        return isNewFixedStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                JSONObject err = value.getJSONObject("err");
                JSONObject start = value.getJSONObject("start");
                JSONObject page = value.getJSONObject("page");
                JSONObject common = value.getJSONObject("common");
                Long ts = value.getLong("ts");

                if (err != null) {
                    // 当前数据中存在报错信息，收集报错信息是给开发人员看的，不需要其他common等信息
                    ctx.output(errorTag, err.toJSONString());
                    value.remove("err");
                }

                if (start != null) {
                    // 当前是启动日志
                    // 注意:输出的是完整的日志信息，不只是start部分信息
                    ctx.output(startTag, value.toJSONString());
                } else if (page != null) {
                    // action,display在page日志中
                    JSONArray displays = value.getJSONArray("displays"); //  "displays": [] 数组
                    //     for (Object display : displays) {
                    //         这样循环的时候，数组中元素类型就丢了，我们要的是String
                    //     }

                    if (displays != null && displays.size() > 0) {
                        for (int i = 0; i < displays.size(); i++) {
                            // 数组中一个元素就是json对象
                            JSONObject display = displays.getJSONObject(i);
                            //      {
                            //        "display_type": "query",
                            //        "item": "35",
                            //        "item_type": "sku_id",
                            //        "order": 1,
                            //        "pos_id": 1
                            //       }
                            //                            // 补充公共信息
                            display.put("page", page);
                            display.put("common", common);
                            display.put("ts", ts);
                            ctx.output(displayTag, display.toJSONString());
                        }
                        // 处理完记得从value中删掉displays数据
                        value.remove("displays");
                    }


                    JSONArray actions = value.getJSONArray("actions"); //  "actions": [] 数组
                    if (actions != null && actions.size() > 0) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            action.put("page", page);
                            action.put("common", common);
                            action.put("ts", ts);
                            ctx.output(actionTag, action.toJSONString());
                        }
                        // 处理完记得从value中删掉actions数据
                        value.remove("actions");

                    }
                } else {
                    // 留空
                }
                // 此时，value中没有actions和displays数据了，只剩page，common,ts
                out.collect(value.toJSONString());

            }
        });
    }

    public SingleOutputStreamOperator<JSONObject> isNewFix(KeyedStream<JSONObject, String> keyedStream) {
        return keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            private ValueState<String> firstLoginState;

            //生命周期，只调用一次
            @Override
            public void open(Configuration parameters) throws Exception {
                firstLoginState = getRuntimeContext().getState(new ValueStateDescriptor<String>("first_login_dt", String.class));
            }

            @Override
            public void processElement(JSONObject jsonObject, Context ctx, Collector<JSONObject> out) throws Exception {
                // 1. 获取当前数据的is_new字段
                JSONObject common = jsonObject.getJSONObject("common");
                String isNew = jsonObject.getJSONObject("common").getString("is_new");
                Long ts = jsonObject.getLong("ts");
                String curDate = DateFormatUtil.tsToDate(ts);

                // 从状态取值
                String firstLoginDate = firstLoginState.value();

                if ("1".equals(isNew)) {

                    if (firstLoginDate != null && firstLoginDate.equals(curDate)) {
                        // 新客户，比对状态中存储的首次登录日期和当前数据的数是否一致，如果不一致，则不是首次登录，不是新客户
                        common.put("is_new", 0);

                    } else if (firstLoginDate == null) {
                        // 新访客，首次登录
                        firstLoginState.update(curDate);
                    } else {
                        // firstLoginDate != null && !firstLoginDate.equals(curDate)
                        // 新访客，首次登录后多次登录

                    }
                } else if ("0".equals(isNew)) {
                    if (firstLoginDate == null) {
                        // 老客户，flink实时数仓里面还没有记录该客户，需补充该客户首次登录日期
                        firstLoginState.update(DateFormatUtil.tsToDate(ts - 24 * 60 * 60 * 1000L));
                    } else {
                        // 任何if中都要有一个else{}收尾，所有条件都要考虑到
                    }
                } else {
                    // 当前数据的is_new不为1，也不为0
                }
                out.collect(jsonObject);
            }
        });
    }

    /**
     * Type parameters:
     * <T> – The type of the elements in the Keyed Stream.
     * <KEY> – The type of the key in the Keyed Stream.
     */
    public KeyedStream<JSONObject, String> getKeyedStreamWithWatermarks(SingleOutputStreamOperator<JSONObject> jsonObjStream) {
        return jsonObjStream
                // 注册水位线
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3l)) // flink内置的乱序水位线策略，乱序等待时间为3秒
                                // 从数据流中提取时间戳
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("ts");
                                    }
                                })
                ).keyBy(new KeySelector<JSONObject, String>() {
                    @Override
                    public String getKey(JSONObject value) throws Exception {
                        return value.getJSONObject("common").getString("mid");
                    }
                });
    }

    public SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream.flatMap(new RichFlatMapFunction<String, JSONObject>() {

            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {

                // 转成JSONObject时报错的话，数据有问题，过滤掉
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    JSONObject page = jsonObject.getJSONObject("page");
                    JSONObject start = jsonObject.getJSONObject("start");
                    JSONObject common = jsonObject.getJSONObject("common");
                    Long ts = jsonObject.getLong("ts");
                    if (page != null || start != null) {

                        // 确保keyBy时，不会出现key为空的情况
                        if (common != null && common.getString("mid") != null && ts != null) {
                            out.collect(jsonObject);
                        }
                    }

                } catch (Exception e) {
                    log.info("过滤掉脏数据" + value);
                }
            }
        });
    }
}
